package com.ln.kafka.v2_4_0.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class AdminSimple {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        AdminClient adminClient = AdminSimple.adminClient();
//        System.out.println(adminClient);
        // 创建Topic实例
//        createTopic();
        // 创建一个Topic，并配置副本集
        createTopicSetReplica();


        // 删除Topic列表
//        delTopics();

        // Topic列表查询
//        topicList();
//        topicListOption();

        // Topic 描述
        describeTopics();

        // 修改Config
//        alterConfig();
//        alterConfigNew();
        // 插叙修改Config
//        describeConfig();

        // 增加partition数量, 在topic描述中查询
//        incrPartitions(2);


    }

    public static String TOPIC_NAME = "topic-3";

    /**
     * 配置副本集的主题
     */
    public static String TOPIC_REPLICA = "replica-test";

    /**
     * 增加partitions数量
     */
    public static void incrPartitions(int partitions) throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(partitions);
        newPartitionsMap.put(TOPIC_NAME, newPartitions);
        CreatePartitionsResult partitionsResult = adminClient.createPartitions(newPartitionsMap);
        partitionsResult.all().get();
    }


    /**
     * 查询Topic描述
     * name: topic-2 ,
     * desc: (name=topic-2,
     *  internal=false,
     *      partitions=(
     *              partition=0,
     *              leader=localhost:9092 (id: 0 rack: null),
     *              replicas=localhost:9092 (id: 0 rack: null),
     *              isr=localhost:9092 (id: 0 rack: null)),
     *              authorizedOperations=[])
     */
    public static void describeTopics() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_REPLICA));
        Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
        stringTopicDescriptionMap.entrySet().forEach(x -> {
            System.out.println("name: " + x.getKey() + " , desc: " + x.getValue());
        });
    }

    /**
     * 修改配置信息: 使用新版
     * 对于非集群版的支持非常的差
     */
    public static void alterConfigNew() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        // 组织两个参数
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
        AlterConfigOp alterConfigOp = new AlterConfigOp(new ConfigEntry("preallocate", "false"), AlterConfigOp.OpType.SET);
        configs.put(configResource, Arrays.asList(alterConfigOp));
        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(configs);
        alterConfigsResult.all().get();
    }

    /**
     * 修改配置信息: 使用旧版
     */
    public static void alterConfig() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        // 组织两个参数
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        Map<ConfigResource, Config> configs = new HashMap<>();
        Config config = new Config(Arrays.asList(new ConfigEntry("preallocate", "true")));
        configs.put(configResource, config);
        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configs);
        alterConfigsResult.all().get();
    }

    /**
     * 查询Topic配置项
     * name: ConfigResource(type=TOPIC, name='topic-2') ,
     * config: Config(
     *  entries=[
     *      ConfigEntry(
     *          name=compression.type,
     *          value=producer,
     *          source=DEFAULT_CONFIG,
     *          isSensitive=false,
     *          isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=leader.replication.throttled.replicas, value=, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=message.downconversion.enable, value=true, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=min.insync.replicas, value=1, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=segment.jitter.ms, value=0, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=cleanup.policy, value=delete, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=flush.ms, value=9223372036854775807, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=follower.replication.throttled.replicas, value=, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=segment.bytes, value=1073741824, source=STATIC_BROKER_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=retention.ms, value=604800000, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=flush.messages, value=9223372036854775807, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=message.format.version, value=2.4-IV1, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=file.delete.delay.ms, value=60000, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=max.compaction.lag.ms, value=9223372036854775807, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=max.message.bytes, value=1000012, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=min.compaction.lag.ms, value=0, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=message.timestamp.type, value=CreateTime, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=preallocate, value=false, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=min.cleanable.dirty.ratio, value=0.5, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=index.interval.bytes, value=4096, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=unclean.leader.election.enable, value=false, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=retention.bytes, value=-1, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=delete.retention.ms, value=86400000, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=segment.ms, value=604800000, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=message.timestamp.difference.max.ms, value=9223372036854775807, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[]),
     *          ConfigEntry(name=segment.index.bytes, value=10485760, source=DEFAULT_CONFIG, isSensitive=false, isReadOnly=false, synonyms=[])])
     */
    public static void describeConfig() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        Map<ConfigResource, Config> configResourceConfigMap = describeConfigsResult.all().get();
        configResourceConfigMap.entrySet().forEach(x -> {
            System.out.println("name: " + x.getKey() + " , config: " + x.getValue());
        });
    }

    /**
     * 删除Topic列表
     *
     */
    public static void delTopics() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
        deleteTopicsResult.all().get();
    }

    /**
     * 常用API：Topic列表查询
     */
    public static void topicList() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        Set<String> names = listTopicsResult.names().get();
        names.stream().forEach(System.out::println);
    }

    /**
     * 常用API：Topic列表查询，加参数
     */
    public static void topicListOption() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSimple.adminClient();
        // 是否查看 internal 选项
        ListTopicsOptions options = new ListTopicsOptions();
        options.listInternal(true);
        ListTopicsResult listTopicsResult = adminClient.listTopics(options);
        Set<String> names = listTopicsResult.names().get();
        names.stream().forEach(System.out::println);
        // 打印 topicListings
        Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        topicListings.forEach(System.out::println);
    }

    /**
     * 常用API：创建一个Topic
     */
    public static void createTopic() {
        AdminClient adminClient = AdminSimple.adminClient();
        // 副本因子
        Short rs = 1;
        NewTopic newTopic = new NewTopic(TOPIC_NAME, /*具体需要多少个partition*/2, rs);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        System.out.println(topics);
    }

    /**
     * 常用API：创建一个Topic，并配置副本集
     * 创建成功之后可以看到对应的信息
         desc:
         (name=replica-test, internal=false, partitions=
         (partition=0, leader=localhost:9095
         (id: 3 rack: null), replicas=localhost:9095
         (id: 3 rack: null), localhost:9093
         (id: 1 rack: null), localhost:9094
         (id: 2 rack: null), isr=localhost:9095
         (id: 3 rack: null), localhost:9093
         (id: 1 rack: null), localhost:9094
         (id: 2 rack: null)),
         (partition=1, leader=localhost:9093
         (id: 1 rack: null), replicas=localhost:9093
         (id: 1 rack: null), localhost:9094
         (id: 2 rack: null), localhost:9095
         (id: 3 rack: null), isr=localhost:9093
         (id: 1 rack: null), localhost:9094
         (id: 2 rack: null), localhost:9095
         (id: 3 rack: null)),
         (partition=2, leader=localhost:9094
         (id: 2 rack: null), replicas=localhost:9094
         (id: 2 rack: null), localhost:9095
         (id: 3 rack: null), localhost:9093
         (id: 1 rack: null), isr=localhost:9094
         (id: 2 rack: null), localhost:9095
         (id: 3 rack: null), localhost:9093
         (id: 1 rack: null)), authorizedOperations=[])
     *
     */
    public static void createTopicSetReplica() {
        AdminClient adminClient = AdminSimple.adminClient();
        // 副本因子
        Short rs = 3;
        NewTopic newTopic = new NewTopic(TOPIC_REPLICA, /*具体需要多少个partition*/3, rs);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        System.out.println(topics);
    }

    /**
     * 常用API：创建一个kafka的AdminClient
     */
    public static AdminClient adminClient() {
        // 准备配置
        Properties properties = new Properties();
        // 配置：kafka服务地址
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093");
        // 创建 adminClient
        AdminClient adminClient = AdminClient.create(properties);
        return adminClient;
    }

}
